package com.lujieni.kafka.controller;

import com.alibaba.fastjson.JSON;
import com.lujieni.kafka.dto.OrderDTO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RefreshScope
public class SendController {


    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 回调的第一种写法
     * 实体类全部转为json格式的字符串最好
     * @param
     */
    @PostMapping("/kafka/send")
    public void sendCallbackOneMessage(@RequestBody OrderDTO orderDTO) {
        // 使用KafkaTemplate发送消息到指定的主题
        kafkaTemplate.send("idea", 2,"a", JSON.toJSONString(orderDTO)).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            /**
             * 当消息发送失败时执行的操作
             * @param ex 发送消息过程中遇到的异常
             */
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败2："+ ex.getMessage());
            }

            /**
             * 当消息发送成功时执行的操作
             * @param result 包含发送结果的对象，包括消息的元数据
             */
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });

    }


}
